#include "CSession.hpp"
#include "CServer.hpp"
#include "LogicSystem.hpp"
#include "RedisMgr.hpp"
#include "const.hpp"
#include <boost/asio/buffer.hpp>
#include <boost/asio/detail/socket_ops.hpp>
#include <boost/asio/ip/tcp.hpp>
#include <boost/asio/write.hpp>
#include <boost/system/error_code.hpp>
#include <boost/uuid/random_generator.hpp>
#include <boost/uuid/uuid.hpp>
#include <boost/uuid/uuid_generators.hpp>
#include <boost/uuid/uuid_io.hpp>
#include <cstddef>
#include <cstring>
#include <exception>
#include <functional>
#include <iostream>
#include <memory>
#include <mutex>
#include <string>

MessageNode ::MessageNode(unsigned short msg_id, unsigned short msg_len,
                          const char *data)
    : _msg_id(msg_id), _msg_len(msg_len) {
  std::memset(_data, 0, MAX_LENGTH + 1);
  std::memcpy(_data, data, HEAD_TOTAL_LEN + msg_len);
}

CSession::CSession(boost::asio::io_context &io_context, CServer *server)
    : _socket(io_context), _server(server), _b_close(false),
      _b_head_parse(false) {
  boost::uuids::uuid a_uuid = boost::uuids::random_generator()();
  _sessionId = boost::uuids::to_string(a_uuid);
  _recv_head_node = std::make_shared<MessageNode>();
}

CSession::~CSession() { std::cout << "~CSession destruct." << std::endl; }

boost::asio::ip::tcp::socket &CSession::GetSocket() { return _socket; }

std::string &CSession::GetSessionId() { return _sessionId; }

void CSession::SetUserId(int uid) { _userId = uid; }

int CSession::GetUserId() { return _userId; }

void CSession::Start() {
  // 读取头部4字节数据
  AsyncReadHead(HEAD_TOTAL_LEN);
}

void CSession::AsyncReadHead(int head_total_len) {
  auto self = shared_from_this();
  asyncReadFull(head_total_len, [self, this, head_total_len](
                                    const boost::system::error_code &ec,
                                    std::size_t bytes_transfered) {
    try {
      if (ec) {
        std::cerr << "AsyncReadHead : handle read failed, error is "
                  << ec.message() << std::endl;
        Close();
        return;
      }
      if (bytes_transfered != head_total_len) {
        std::cout << "read length not match, read [" << bytes_transfered
                  << "],total[" << head_total_len << "]" << std::endl;
        Close();
        return;
      }

      short msg_id = 0;
      memcpy(&msg_id, _data, HEAD_ID_LEN);
      // 网路字节序转换为本地字节序
      msg_id = boost::asio::detail::socket_ops::network_to_host_short(msg_id);
      std::cout << "msg_id is " << msg_id << std::endl;
      // id非法
      if (msg_id > MAX_LENGTH) {
        std::cerr << "Invalid msg_id is " << msg_id << std::endl;
        Close();
        return;
      }
      _recv_head_node->_msg_id = msg_id;
      short msg_len = 0;
      memcpy(&msg_len, _data + HEAD_ID_LEN, HEAD_DATA_LEN);
      msg_len = boost::asio::detail::socket_ops::network_to_host_short(msg_len);
      std::cout << "msg_len is " << msg_len << std::endl;
      // 长度非法
      if (msg_len > MAX_LENGTH) {
        std::cerr << "Invalid msg_len is " << msg_len << std::endl;
        Close();
        return;
      }
      _recv_head_node->_msg_len = msg_len;
      // 接收消息体
      AsyncReadBody(msg_len);

    } catch (std::exception &e) {
      std::cerr << "Exception: " << e.what() << std::endl;
      Close();
    }
  });
}

// 读取完整长度
void CSession::asyncReadFull(
    std::size_t maxlength,
    std::function<void(const boost::system::error_code &ec,
                       std::size_t bytes_transfered)>
        handler) {
  std::memset(_data, 0, MAX_LENGTH + 1);
  asyncReadLen(0, maxlength, handler);
}

// 读取指定字节数
void CSession::asyncReadLen(
    std::size_t read_len, std::size_t total_len,
    std::function<void(const boost::system::error_code &ec,
                       std::size_t bytes_transfered)>
        handler) {
  auto self = shared_from_this();
  _socket.async_read_some(
      boost::asio::buffer(_data + read_len, total_len - read_len),
      [read_len, total_len, handler, self](const boost::system::error_code &ec,
                                           std::size_t bytes_transfered) {
        if (ec) {
          std::cout << "asyncReadLen: error is " << ec.message() << std::endl;
          // 出现错误，调用回调函数
          handler(ec, read_len + bytes_transfered);
          return;
        }
        // 读取到所有数据
        if (read_len + bytes_transfered == total_len) {
          handler(ec, read_len + bytes_transfered);
        } else {
          // 长度不足 继续读取
          self->asyncReadLen(read_len + bytes_transfered, total_len, handler);
        }
      });
}

void CSession::Send(const char *msg, unsigned short msg_len,
                    unsigned short msg_id) {
  std::lock_guard<std::mutex> lock(_send_mutex);
  int send_que_size = _send_que.size();
  if (send_que_size > MAX_SENDQUE) {
    std::cerr << "msg_id: " << msg_id << " send que fulled, size is "
              << send_que_size << std::endl;
    return;
  }
  char all_msg[MAX_LENGTH + 1];
  std::memset(all_msg, 0, MAX_LENGTH + 1);
  unsigned short _msg_id =
      boost::asio::detail::socket_ops::host_to_network_short(msg_id);
  unsigned short _msg_len =
      boost::asio::detail::socket_ops::host_to_network_short(msg_len);
  std::memcpy(all_msg, &_msg_id, HEAD_ID_LEN);
  std::memcpy(all_msg + HEAD_ID_LEN, &_msg_len, HEAD_DATA_LEN);
  std::memcpy(all_msg + HEAD_TOTAL_LEN, msg, msg_len);
  std::cout << "msg_id " << msg_id << ": send msg length is " << msg_len
            << " . msg is " << all_msg + HEAD_TOTAL_LEN << std::endl;
  _send_que.push(std::make_shared<MessageNode>(msg_id, msg_len, all_msg));
  // 之前队列有数据
  if (send_que_size > 0) {
    return;
  }
  auto &msgNode = _send_que.front();

  boost::asio::async_write(
      _socket,
      boost::asio::buffer(msgNode->_data, HEAD_TOTAL_LEN + msgNode->_msg_len),
      std::bind(&CSession::HandleWrite, this, shared_from_this(),
                std::placeholders::_1, std::placeholders::_2));
}

void CSession::HandleWrite(std::shared_ptr<CSession> self,
                           const boost::system::error_code &ec,
                           std::size_t bytes_transferred) {
  try {
    // 写数据出错
    if (ec) {
      std::cerr << "send message failed!  error is " << ec.message()
                << std::endl;
      self->Close();
      return;
    } else {
      std::unique_lock<std::mutex> lock(_send_mutex);
      self->_send_que.pop();
      // 消息队列中还存在消息，就继续发送数据
      if (!self->_send_que.empty()) {
        auto &msgNode = _send_que.front();
        boost::asio::async_write(
            _socket, boost::asio::buffer(msgNode->_data, msgNode->_msg_len),
            std::bind(&CSession::HandleWrite, this, shared_from_this(),
                      std::placeholders::_1, std::placeholders::_2));
      }
    }
  } catch (std::exception &e) {
    std::cout << "CSession::HandleWrite failed : " << e.what() << std::endl;
    Close();
    return;
  }
}

void CSession::AsyncReadBody(int body_len) {
  auto self = shared_from_this();
  asyncReadFull(body_len, [self, this,
                           body_len](const boost::system::error_code &ec,
                                     std::size_t bytes_transfered) {
    try {
      if (ec) {
        std::cerr << "AsyncReadBody : handle read failed, error is "
                  << ec.message() << std::endl;
        Close();
        _server->ClearSession(_sessionId);
        return;
      }
      if (bytes_transfered != body_len) {
        std::cerr << "read length not match, read [" << bytes_transfered
                  << "], total [" << body_len << "]" << std::endl;
        Close();
        _server->ClearSession(_sessionId);
        return;
      }
      memset(_recv_head_node->_data, 0, MAX_LENGTH + 1);
      memcpy(_recv_head_node->_data, _data, body_len);
      // _recv_head_node->_data[body_len + 1] = 0;
      std::cout << "receive data is " << _recv_head_node->_data << std::endl;
      // 将消息投递到逻辑队列中
      auto node = std::make_shared<MessageNode>();
      node->_msg_len = _recv_head_node->_msg_len;
      node->_msg_id = _recv_head_node->_msg_id;
      std::memcpy(node->_data, _recv_head_node->_data,
                  _recv_head_node->_msg_len);
      LogicSystem::GetInstance().PostMsgToQue(
          std::make_shared<LogicNode>(shared_from_this(), node));
      // 继续监听头部接收事件
      if (_b_close) {

        return;
      }
      AsyncReadHead(HEAD_TOTAL_LEN);
    } catch (std::exception &e) {
      std::cerr << e.what() << std::endl;
    }
  });
}

void CSession::Close() {
  std::cout << "Close: " << _userId << std::endl;
  if (_userId != 0) {
    std::string key = USER_IP_PREFIX + std::to_string(_userId);
    std::cout << "Del " << key << std::endl;
    RedisMgr::GetInstance().Del(key);
  }
  _server->ClearSession(_sessionId);
  _socket.close();
  _b_close = true;
}